/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nifi.processors.huawei.obs;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.json.JsonMapper;
import com.obs.services.ObsClient;
import com.obs.services.exception.ObsException;
import com.obs.services.model.*;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.annotation.behavior.*;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.huawei.abstractprocessor.AbstractOBSProcessor;
import org.apache.nifi.processors.huawei.obs.encryption.StandardOBSEncryptionService;

import java.io.*;
import java.net.URLEncoder;
import java.nio.file.Files;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

import static org.apache.nifi.processors.huawei.common.PropertyDescriptors.*;
import static org.apache.nifi.processors.huawei.obs.Constants.*;

@SupportsBatching
@InputRequirement(Requirement.INPUT_REQUIRED)
@Tags({"HuaweiCloud", "OBS", "Archive", "Put"})
@CapabilityDescription("Puts FlowFiles to an OBS Bucket.\n" +
        "The upload uses either the PutOBSObject method or the PutOBSMultipartUpload method.  The PutOBSObject method " +
        "sends the file in a single synchronous call, but it has a 5GB size limit.  Larger files are sent using the " +
        "PutOBSMultipartUpload method.  This multipart process " +
        "saves state after each step so that a large upload can be resumed with minimal loss if the processor or " +
        "cluster is stopped and restarted.\n" +
        "A multipart upload consists of three steps:\n" +
        "  1) initiate upload,\n" +
        "  2) upload the parts, and\n" +
        "  3) complete the upload.\n" +
        "For multipart uploads, the processor saves state locally tracking the upload ID and parts uploaded, which " +
        "must both be provided to complete the upload.\n" +
        "The HuaweiCloud libraries select an endpoint URL based on the HuaweiCloud region, but this can be overridden with the " +
        "'Endpoint Override URL' property for use with other OBS-compatible endpoints.\n" +
        "The OBS API specifies that the maximum file size for a PutOBSObject upload is 5GB. It also requires that " +
        "parts in a multipart upload must be at least 5MB in size, except for the last part.  These limits " +
        "establish the bounds for the Multipart Upload Threshold and Part Size properties.")
@DynamicProperty(name = "The name of a User-Defined Metadata field to add to the OBS Object",
        value = "The value of a User-Defined Metadata field to add to the OBS Object",
        description = "Allows user-defined metadata to be added to the OBS object as key/value pairs",
        expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
@ReadsAttribute(attribute = "filename", description = "Uses the FlowFile's filename as the filename for the OBS object")
@WritesAttributes({
        @WritesAttribute(attribute = OBS_BUCKET, description = "The OBS bucket where the Object was put in OBS"),
        @WritesAttribute(attribute = OBS_OBJECT, description = "The OBS key within where the Object was put in OBS"),
        @WritesAttribute(attribute = OBS_CONTENT_TYPE, description = "The OBS content type of the OBS Object that put in OBS"),
        @WritesAttribute(attribute = OBS_VERSION, description = "The version of the OBS Object that was put to OBS"),
        @WritesAttribute(attribute = OBS_EXCEPTION, description = "The class name of the exception thrown during processor execution"),
        @WritesAttribute(attribute = OBS_ADDITIONAL_DETAILS, description = "The OBS supplied detail from the failed operation"),
        @WritesAttribute(attribute = OBS_STATUS_CODE, description = "The HTTP error code (if available) from the failed operation"),
        @WritesAttribute(attribute = OBS_ERROR_CODE, description = "The OBS moniker of the failed operation"),
        @WritesAttribute(attribute = OBS_ERROR_Message, description = "The OBS exception message from the failed operation"),
        @WritesAttribute(attribute = OBS_E_TAG, description = "The ETag of the OBS Object"),
        @WritesAttribute(attribute = OBS_CONTENT_DISPOSITION, description = "The content disposition of the OBS Object that put in OBS"),
        @WritesAttribute(attribute = OBS_CACHE_CONTROL, description = "The cache-control header of the OBS Object"),
        @WritesAttribute(attribute = OBS_UPLOAD_ID, description = "The uploadId used to upload the Object to OBS"),
        @WritesAttribute(attribute = OBS_EXPIRATION_TIME, description = "A human-readable form of the expiration date of " +
                "the OBS object, if one is set"),
        @WritesAttribute(attribute = OBS_SSE_ALGORITHM, description = "The server side encryption algorithm of the object"),
        @WritesAttribute(attribute = OBS_USER_META, description = "A human-readable form of the User Metadata of " +
                "the OBS object, if any was set"),
        @WritesAttribute(attribute = OBS_ENCRYPTION_STRATEGY, description = "The name of the encryption strategy, if any was set"),})
public class PutOBSObject extends AbstractOBSProcessor {
    public static final long MIN_PART_SIZE = 50L * 1024L * 1024L;
    public static final long MAX_PUTOBJECT_SIZE = 5L * 1024L * 1024L * 1024L;
    public static final String CONTENT_DISPOSITION_INLINE = "inline";
    public static final String CONTENT_DISPOSITION_ATTACHMENT = "attachment";

    public static final String CONTENT_DISPOSITION_FILENAME = "filename";

    private static final Set<String> STORAGE_CLASSES = Collections.unmodifiableSortedSet(new TreeSet<>(
            Arrays.stream(StorageClassEnum.values()).map(StorageClassEnum::getCode).collect(Collectors.toSet())
    ));

    public static final PropertyDescriptor CONTENT_TYPE = new PropertyDescriptor.Builder()
            .name("Content Type")
            .displayName("Content Type")
            .description("Sets the Content-Type HTTP header indicating the type of content stored in the associated " +
                    "object. The value of this header is a standard MIME type.\n" +
                    "OBS Java client will attempt to determine the correct content type if one hasn't been set" +
                    " yet. Users are responsible for ensuring a suitable content type is set when uploading streams. If " +
                    "no content type is provided and cannot be determined by the filename, the default content type " +
                    "\"application/octet-stream\" will be used.")
            .required(false)
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor CONTENT_DISPOSITION = new PropertyDescriptor.Builder()
            .name("Content Disposition")
            .displayName("Content Disposition")
            .description("Sets the Content-Disposition HTTP header indicating if the content is intended to be displayed inline or should be downloaded.\n " +
                    "Possible values are 'inline' or 'attachment' or 'filename'. If this property is not specified, object's content-disposition will be set to null. " +
                    "when filename is selected, object's content-disposition will be set to filename" +
                    "When 'attachment' is selected, '; filename=' plus object key are automatically appended to form final value 'attachment; filename=\"filename.jpg\"'.")
            .required(false)
            .allowableValues(CONTENT_DISPOSITION_INLINE, CONTENT_DISPOSITION_ATTACHMENT, CONTENT_DISPOSITION_FILENAME)
            .build();

    public static final PropertyDescriptor CACHE_CONTROL = new PropertyDescriptor.Builder()
            .name("Cache Control")
            .displayName("Cache Control")
            .description("Sets the Cache-Control HTTP header indicating the caching directives of the associated object. Multiple directives are comma-separated.")
            .required(false)
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor STORAGE_CLASS = new PropertyDescriptor.Builder()
            .name("Storage Class")
            .required(true)
            .allowableValues(STORAGE_CLASSES)
            .defaultValue(StorageClassEnum.STANDARD.getCode())
            .build();

    public static final PropertyDescriptor MULTIPART_THRESHOLD = new PropertyDescriptor.Builder()
            .name("Multipart Threshold")
            .description("Specifies the file size threshold for switch from the PutOBSObject API to the " +
                    "PutOBSMultipartUpload API.  Flow files bigger than this limit will be sent using the stateful " +
                    "multipart process.\n" +
                    "The valid range is 50MB to 5GB.")
            .required(true)
            .defaultValue("5 GB")
            .addValidator(StandardValidators.createDataSizeBoundsValidator(MIN_PART_SIZE, MAX_PUTOBJECT_SIZE))
            .build();

    public static final PropertyDescriptor MULTIPART_PART_SIZE = new PropertyDescriptor.Builder()
            .name("Multipart Part Size")
            .description("Specifies the part size for use when the PutOBSMultipart Upload API is used.\n" +
                    "Flow files will be broken into chunks of this size for the upload process, but the last part " +
                    "sent can be smaller since it is not padded.\n" +
                    "The valid range is 50MB to 5GB.")
            .required(true)
            .defaultValue("5 GB")
            .addValidator(StandardValidators.createDataSizeBoundsValidator(MIN_PART_SIZE, MAX_PUTOBJECT_SIZE))
            .build();

    public static final PropertyDescriptor MULTIPART_AGEOFF_INTERVAL = new PropertyDescriptor.Builder()
            .name("Multipart Upload AgeOff Interval")
            .description("Specifies the interval at which existing multipart uploads in HuaweiCloud OBS will be evaluated " +
                    "for ageoff.  When processor is triggered it will initiate the ageoff evaluation if this interval has been " +
                    "exceeded.")
            .required(true)
            .defaultValue("60 min")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .build();

    public static final PropertyDescriptor MULTIPART_MAX_AGE = new PropertyDescriptor.Builder()
            .name("Multipart Upload Max Age Threshold")
            .description("Specifies the maximum age for existing multipart uploads in HuaweiCloud OBS.  When the ageoff " +
                    "process occurs, any upload older than this threshold will be aborted.")
            .required(true)
            .defaultValue("7 days")
            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
            .build();

    public static final PropertyDescriptor MULTIPART_TEMP_DIR = new PropertyDescriptor.Builder()
            .name("obs-temporary-directory-multipart")
            .displayName("Temporary Directory Multipart State")
            .description("Directory in which, for multipart uploads, the processor will locally save the state tracking the upload ID and parts "
                    + "uploaded which must both be provided to complete the upload.")
            .required(true)
            .addValidator(StandardValidators.FILE_EXISTS_VALIDATOR)
            .defaultValue("${java.io.tmpdir}")
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();

    public static final List<PropertyDescriptor> properties = Collections.unmodifiableList(
            Arrays.asList(
                    OBS_REGION,
                    ENDPOINT_OVERRIDE_URL,
                    BUCKET,
                    ACCESS_KEY,
                    SECRET_KEY,
                    HUAWEI_CREDENTIALS_PROVIDER_SERVICE,
                    KEY,
                    STORAGE_CLASS,
                    CONTENT_TYPE,
                    CONTENT_DISPOSITION,
                    CACHE_CONTROL,
                    TIMEOUT,
                    OWNER,
                    READ_USER_LIST,
                    FULL_CONTROL_USER_LIST,
                    READ_ACL_LIST,
                    WRITE_ACL_LIST,
                    CANNED_ACL,
                    MULTIPART_THRESHOLD,
                    MULTIPART_PART_SIZE,
                    MULTIPART_AGEOFF_INTERVAL,
                    MULTIPART_MAX_AGE,
                    MULTIPART_TEMP_DIR,
                    ENCRYPTION_SERVICE,
                    PROXY_CONFIGURATION_SERVICE
                    ));

    private volatile String tempDirMultipart = System.getProperty("java.io.tmpdir");

    @OnScheduled
    public void setTempDir(final ProcessContext context) {
        this.tempDirMultipart = context.getProperty(MULTIPART_TEMP_DIR).evaluateAttributeExpressions().getValue();
    }

    @Override
    public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return properties;
    }

    @Override
    public PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
        return new PropertyDescriptor.Builder()
                .name(propertyDescriptorName)
                .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
                .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
                .dynamic(true)
                .build();
    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) {
        FlowFile flowFile = session.get();
        if (flowFile == null) {
            return;
        }
        try {
        final long startNanos = System.nanoTime();
        CustomProperties customProperties = new CustomProperties();
        customProperties.setBucket(context.getProperty(BUCKET).evaluateAttributeExpressions(flowFile).getValue());
        customProperties.setObjectKey(context.getProperty(KEY).evaluateAttributeExpressions(flowFile).getValue());
        customProperties.setCacheKey(getIdentifier() + "/" + customProperties.getBucket() + "/" + customProperties.getObjectKey());
        customProperties.setFileName(flowFile.getAttributes().get(CoreAttributes.FILENAME.key()));
        customProperties.setMultipartThreshold(context.getProperty(MULTIPART_THRESHOLD).asDataSize(DataUnit.B).longValue());
        customProperties.setMultipartPartSize(context.getProperty(MULTIPART_PART_SIZE).asDataSize(DataUnit.B).longValue());
        customProperties.setContentDisposition(context.getProperty(CONTENT_DISPOSITION).getValue());
        customProperties.setContentType(context.getProperty(CONTENT_TYPE).evaluateAttributeExpressions(flowFile).getValue());
        customProperties.setCacheControl(context.getProperty(CACHE_CONTROL).evaluateAttributeExpressions(flowFile).getValue());
        customProperties.setEncryptionService(context.getProperty(ENCRYPTION_SERVICE).asControllerService(ObsServiceEncryptionService.class));
        if (customProperties.getEncryptionService() == null) {
            customProperties.setEncryptionService(new StandardOBSEncryptionService());
        }
        customProperties.setStorageClass(StorageClassEnum.valueOf(context.getProperty(STORAGE_CLASS).getValue()));
        customProperties.setFlowFile(flowFile);

        final Map<String, String> attributes = new HashMap<>();
        attributes.put(OBS_BUCKET, customProperties.getBucket());
        attributes.put(OBS_OBJECT, customProperties.getObjectKey());

        final long now = System.currentTimeMillis();

        ageOffUploads(context, client, now, customProperties.getBucket());
        session.read(customProperties.getFlowFile(), new InputStreamCallback() {
            @Override
            public void process(final InputStream rawIn) throws IOException {
                attributes.putAll(doUpload(customProperties, rawIn, client, context));
            }
        });

        if (!attributes.isEmpty()) {
            flowFile = session.putAllAttributes(flowFile, attributes);
        }
        session.transfer(flowFile, REL_SUCCESS);
        final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
        session.getProvenanceReporter().send(flowFile, attributes.get(OBJECT_URL), millis);
        getLogger().info("Successfully put {} to OBS in {} milliseconds", customProperties.getFlowFile(), millis);
        try {
            removeLocalState(customProperties.getCacheKey());
        } catch (IOException e) {
            getLogger().info("Error trying to delete key {} from cache: {}",
                    customProperties.getCacheKey(), e.getMessage());
        }
        } catch (final ProcessException | ObsException pe) {
            extractExceptionDetails(pe, session, flowFile);
            getLogger().error("Failed to put {} to Huawei OBS due to {}", new Object[]{flowFile, pe});
            flowFile = session.penalize(flowFile);
            session.transfer(flowFile, REL_FAILURE);
        }
    }

    public File getPersistenceFile() {
        return new File(this.tempDirMultipart + File.separator + getIdentifier());
    }

    protected boolean localUploadExistsInOBS(final ObsClient obsClient, final String bucket, final MultipartState localState) {
        ListMultipartUploadsRequest listRequest = new ListMultipartUploadsRequest(bucket);
        MultipartUploadListing listing = obsClient.listMultipartUploads(listRequest);
        for (MultipartUpload upload : listing.getMultipartTaskList()) {
            if (upload.getUploadId().equals(localState.getUploadId())) {
                return true;
            }
        }
        return false;
    }

    public synchronized MultipartState getLocalStateIfInOBS(final ObsClient obsClient, final String bucket,
                                                            final String objectKey) throws IOException {
        MultipartState currState = getLocalState(objectKey);
        if (currState == null) {
            return null;
        }

        if (localUploadExistsInOBS(obsClient, bucket, currState)) {
            getLogger().info("Local state for {} loaded with uploadId {} and {} partETags",
                    objectKey, currState.getUploadId(), currState.getPartETags().size());
            return currState;
        } else {
            getLogger().info("Local state for {} with uploadId {} does not exist in OBS, deleting local state",
                    objectKey, currState.getUploadId());
            persistLocalState(objectKey, null);
            return null;
        }
    }

    protected synchronized MultipartState getLocalState(final String objectKey) throws IOException {
        // get local state if it exists
        final File persistenceFile = getPersistenceFile();

        if (persistenceFile.exists()) {
            final Properties props = new Properties();
            try (final FileInputStream fis = new FileInputStream(persistenceFile)) {
                props.load(fis);
            } catch (IOException ioe) {
                getLogger().warn("Failed to recover local state for {} due to {}. Assuming no local state and " +
                        "restarting upload.", objectKey, ioe.getMessage());
                return null;
            }
            if (props.containsKey(objectKey)) {
                final String localSerialState = props.getProperty(objectKey);
                if (localSerialState != null) {
                    try {
                        return MultipartState.newMultipartState(localSerialState);
                    } catch (final RuntimeException rte) {
                        getLogger().warn("Failed to recover local state for {} due to corrupt data in state.", objectKey, rte.getMessage());
                        return null;
                    }
                }
            }
        }
        return null;
    }

    public synchronized void persistLocalState(final String objectKey, final MultipartState currState) throws IOException {
        JsonMapper mapper = new JsonMapper();
        final String currStateStr = (currState == null) ? null : mapper.writeValueAsString(currState);
        final File persistenceFile = getPersistenceFile();
        final File parentDir = persistenceFile.getParentFile();
        if (!parentDir.exists() && !parentDir.mkdirs()) {
            throw new IOException("Persistence directory (" + parentDir.getAbsolutePath() + ") does not exist and " +
                    "could not be created.");
        }
        final Properties props = new Properties();
        if (persistenceFile.exists()) {
            try (final FileInputStream fis = new FileInputStream(persistenceFile)) {
                props.load(fis);
            }
        }
        if (currStateStr != null) {
            currState.setTimestamp(System.currentTimeMillis());
            props.setProperty(objectKey, currStateStr);
        } else {
            props.remove(objectKey);
        }

        if (props.size() > 0) {
            try (final FileOutputStream fos = new FileOutputStream(persistenceFile)) {
                props.store(fos, null);
            } catch (IOException ioe) {
                getLogger().error("Could not store state {} due to {}.",
                        persistenceFile.getAbsolutePath(), ioe.getMessage());
            }
        } else {
            if (persistenceFile.exists()) {
                try {
                    Files.delete(persistenceFile.toPath());
                } catch (IOException ioe) {
                    getLogger().error("Could not remove state file {} due to {}.",
                            persistenceFile.getAbsolutePath(), ioe.getMessage());
                }
            }
        }
    }

    protected synchronized void removeLocalState(final String objectKey) throws IOException {
        persistLocalState(objectKey, null);
    }

    // Clear the multi-segment upload status information in the local cache
    private synchronized void ageOffLocalState(long ageCutoff) {
        // get local state if it exists
        final File persistenceFile = getPersistenceFile();
        if (!persistenceFile.exists()) {
            return;
        }
        Properties props = new Properties();
        try (final FileInputStream fis = new FileInputStream(persistenceFile)) {
            props.load(fis);
        } catch (final IOException ioe) {
            getLogger().warn("Failed to ageoff remove local state due to {}",
                    ioe.getMessage());
            return;
        }
        for (Entry<Object, Object> entry : props.entrySet()) {
            final String key = (String) entry.getKey();
            final String localSerialState = props.getProperty(key);
            if (StringUtils.isBlank(localSerialState)) {
                continue;
            }
            final MultipartState state;
            try {
                state = MultipartState.newMultipartState(localSerialState);
            } catch (JsonProcessingException e) {
                getLogger().warn("Failed to recover local state for {} due to corrupt data in state.", key, e.getMessage());
                continue;
            }
            if (state.getTimestamp() < ageCutoff) {
                getLogger().warn("Removing local state for {} due to exceeding ageoff time",
                        key);
                try {
                    removeLocalState(key);
                } catch (final IOException ioe) {
                    getLogger().warn("Failed to remove local state for {} due to {}",
                            key, ioe.getMessage());

                }
            }
        }
    }


    private Map<String, String> doUpload(CustomProperties customProperties, InputStream rawIn, ObsClient obsClient, ProcessContext context) throws IOException {
        Map<String, String> attributes = new HashMap<>();
        try (final InputStream in = new BufferedInputStream(rawIn)) {
            customProperties.setContent(in);
            final ObjectMetadata objectMetadata = new ObjectMetadata();
            objectMetadata.setContentLength(customProperties.getFlowFile().getSize());
            if (StringUtils.isNoneBlank(customProperties.getContentType())) {
                objectMetadata.setContentType(customProperties.getContentType());
                attributes.put(OBS_CONTENT_TYPE, customProperties.getContentType());
            }
            if (StringUtils.isNoneBlank(customProperties.getCacheControl())) {
                objectMetadata.setCacheControl(customProperties.getCacheControl());
                attributes.put(OBS_CACHE_CONTROL, customProperties.getCacheControl());
            }

            final String contentDisposition = customProperties.getContentDisposition();
            String dispositionValue = null;
            if (CONTENT_DISPOSITION_INLINE.equals(contentDisposition)) {
                dispositionValue = contentDisposition;
                attributes.put(OBS_CONTENT_DISPOSITION, contentDisposition);
            } else if (CONTENT_DISPOSITION_ATTACHMENT.equals(contentDisposition)) {
                dispositionValue = CONTENT_DISPOSITION_ATTACHMENT + "; filename=\"" + dispositionValue + "\"";
                attributes.put(OBS_CONTENT_DISPOSITION, dispositionValue);
            } else if (CONTENT_DISPOSITION_FILENAME.equals(contentDisposition)) {
                dispositionValue = URLEncoder.encode(customProperties.getFileName(), "UTF-8");
            }
            if (StringUtils.isNotBlank(dispositionValue)) {
                objectMetadata.setContentDisposition(dispositionValue);
            }
            Map<String, Object> userMetaData = buildUserMetaData(context, customProperties);
            if (MapUtils.isNotEmpty(userMetaData)) {
                objectMetadata.setUserMetadata(userMetaData);
            }
            // Whether to encrypt it and how to encrypt it
            attributes.put(OBS_ENCRYPTION_STRATEGY, customProperties.getEncryptionService().getStrategyName());
            if (ObsServiceEncryptionService.STRATEGY_NAME_SSE_KMS.equals(customProperties.getEncryptionService().getStrategyName())) {
                // sseKms
                attributes.put(OBS_SSE_ALGORITHM, SSEAlgorithmEnum.KMS.getCode());
            } else if (ObsServiceEncryptionService.STRATEGY_NAME_SSE_C.equals(customProperties.getEncryptionService().getStrategyName())) {
                // sseC
                attributes.put(OBS_SSE_ALGORITHM, SSEAlgorithmEnum.AES256.getCode());
            }
            if (customProperties.getFlowFile().getSize() <= customProperties.getMultipartThreshold()) {
                singleUpload(customProperties, attributes, context, objectMetadata, obsClient);
            } else {
                multipartUpload(customProperties, attributes, context, objectMetadata, obsClient);
            }
        }
        return attributes;
    }

    private void singleUpload(final CustomProperties customProperties, Map<String, String> attributes, ProcessContext context, ObjectMetadata objectMetadata, ObsClient obsClient) {

        final PutObjectRequest request = new PutObjectRequest(customProperties.getBucket(), customProperties.getObjectKey(), customProperties.getContent());
        request.setMetadata(objectMetadata);
        customProperties.getEncryptionService().configurePutObjectRequest(request, objectMetadata);
        request.getMetadata().setObjectStorageClass(customProperties.getStorageClass());
        // 设置acl
        setAcl(customProperties, request, context);
        try {
            final PutObjectResult result = obsClient.putObject(request);
            PutResult putResult = new PutResult();
            putResult.setBucketName(result.getBucketName());
            putResult.setETag(result.getEtag());
            putResult.setObjectUrl(result.getObjectUrl());
            putResult.setStorageClass(result.getObjectStorageClass());
            putResult.setVersionId(result.getVersionId());
            putResult.setObjectKey(result.getObjectKey());
            putResult.setPutType(OBS_API_METHOD_PUT_OBJECT);
            attributes.putAll(getResultAttributes(putResult, objectMetadata));
        } catch (Exception e) {
            getLogger().info("Failure completing upload flowfile={} bucket={} key={} reason={}",
                    customProperties.getFileName(), customProperties.getBucket(), customProperties.getObjectKey(), e.getMessage());
            throw (e);
        }
    }

    private void multipartUpload(final CustomProperties customProperties, Map<String, String> attributes, ProcessContext context, ObjectMetadata objectMetadata, ObsClient obsClient) throws IOException {
        final String bucket = customProperties.getBucket();
        final String objectKey = customProperties.getObjectKey();
        final String fileName = customProperties.getFileName();

        MultipartState currentState = initiateMultipartUpload(customProperties, objectMetadata, obsClient, context);
        attributes.put(OBS_UPLOAD_ID, currentState.getUploadId());
        // upload the rest parts
        uploadAllParts(currentState, customProperties, objectMetadata, obsClient);

        // complete multipart upload
        CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(
                bucket, objectKey, currentState.getUploadId(), currentState.getPartETags());

        // No call to an encryption service is needed for a CompleteMultipartUploadRequest.
        try {
            CompleteMultipartUploadResult completeResult = obsClient.completeMultipartUpload(completeRequest);
            getLogger().info("Success completing upload flowfile={} etag={} uploadId={}",
                    fileName, completeResult.getEtag(), currentState.getUploadId());
            PutResult putResult = new PutResult();
            putResult.setBucketName(completeResult.getBucketName());
            putResult.setETag(completeResult.getEtag());
            putResult.setObjectUrl(completeResult.getObjectUrl());
            putResult.setStorageClass(currentState.getStorageClass());
            putResult.setVersionId(completeResult.getVersionId());
            putResult.setObjectKey(completeResult.getObjectKey());
            putResult.setPutType(OBS_API_METHOD_MULTIPART_UPLOAD);
            attributes.putAll(getResultAttributes(putResult, objectMetadata));
        } catch (Exception e) {
            getLogger().info("Failure completing upload flowfile={} bucket={} key={} reason={}",
                    fileName, bucket, objectKey, e.getMessage());
            throw (e);
        }
    }

    public MultipartState initiateMultipartUpload(CustomProperties customProperties, ObjectMetadata objectMetadata, ObsClient obsClient, ProcessContext context) throws IOException {
        final String fileName = customProperties.getFileName();
        final String bucket = customProperties.getBucket();
        final String objectKey = customProperties.getObjectKey();

        // multipart upload
        // load or create persistent state
        MultipartState currentState = getState(obsClient, customProperties);
        // initiate multipart upload or find position in file
        if (StringUtils.isBlank(currentState.getUploadId())) {
            final InitiateMultipartUploadRequest initiateRequest = new InitiateMultipartUploadRequest(bucket, objectKey);
            initiateRequest.setMetadata(objectMetadata);
            customProperties.getEncryptionService().configureInitiateMultipartUploadRequest(initiateRequest, objectMetadata);
            initiateRequest.getMetadata().setObjectStorageClass(customProperties.getStorageClass());
            // 设置acl
            setAcl(customProperties, initiateRequest, context);
            try {
                final InitiateMultipartUploadResult initiateResult = obsClient.initiateMultipartUpload(initiateRequest);
                currentState.setUploadId(initiateResult.getUploadId());
                currentState.getPartETags().clear();
                try {
                    persistLocalState(customProperties.getCacheKey(), currentState);
                } catch (Exception e) {
                    getLogger().info("Exception saving cache state while processing flow file: " +
                            e.getMessage());
                    throw (new ProcessException("Exception saving cache state", e));
                }
                getLogger().info("Success initiating upload flowfile={} available={} position={} length={} bucket={} key={} uploadId={}",
                        fileName, customProperties.getContent().available(), currentState.getFilePosition(),
                        currentState.getContentLength(), bucket, objectKey,
                        currentState.getUploadId());

            } catch (Exception e) {
                getLogger().info("Failure initiating upload flowfile={} bucket={} key={} reason={}",
                        customProperties.getFileName(), customProperties.getBucket(), customProperties.getObjectKey(), e.getMessage());
                throw (e);
            }
        } else {
            if (currentState.getFilePosition() > 0) {
                try {
                    final long skipped = customProperties.getContent().skip(currentState.getFilePosition());
                    if (skipped != currentState.getFilePosition()) {
                        getLogger().info("Failure skipping to resume upload flowfile={} bucket={} key={} position={} skipped={}",
                                customProperties, bucket, objectKey,
                                currentState.getFilePosition(), skipped);
                    }
                } catch (Exception e) {
                    getLogger().info("Failure skipping to resume upload flowfile={} bucket={} key={} position={} reason={}",
                            fileName, bucket, objectKey, currentState.getFilePosition(),
                            e.getMessage());
                    throw (new ProcessException(e));
                }
            }
        }
        return currentState;
    }

    private void uploadAllParts(MultipartState currentState, CustomProperties customProperties, ObjectMetadata objectMetadata, ObsClient obsClient) throws IOException {
        // upload parts
        long thisPartSize;
        for (int part = currentState.getPartETags().size() + 1;
             currentState.getFilePosition() < currentState.getContentLength(); part++) {
            if (!PutOBSObject.this.isScheduled()) {
                throw new IOException(OBS_PROCESS_UNSCHEDULED_MESSAGE + " flowfile=" + customProperties.getFileName() + " part=" + part + " uploadId=" + currentState.getUploadId());
            }
            boolean isLastPart = (currentState.getContentLength() - currentState.getFilePosition()) <= currentState.getPartSize();
            thisPartSize = isLastPart ? currentState.getContentLength() - currentState.getFilePosition() : currentState.getPartSize();
            UploadPartRequest uploadRequest = new UploadPartRequest(customProperties.getBucket(), customProperties.getObjectKey());
            uploadRequest.setUploadId(currentState.getUploadId());
            uploadRequest.setInput(customProperties.getContent());
            uploadRequest.setPartNumber(part);
            uploadRequest.setPartSize(thisPartSize);
            // The last section of the file is set to automatically close the stream
            uploadRequest.setAutoClose(isLastPart);
            customProperties.getEncryptionService().configureUploadPartRequest(uploadRequest, objectMetadata);
            try {
                UploadPartResult uploadPartResult = obsClient.uploadPart(uploadRequest);
                currentState.addPartETag(new PartEtag(uploadPartResult.getEtag(), uploadPartResult.getPartNumber()));
                currentState.setFilePosition(currentState.getFilePosition() + thisPartSize);
                try {
                    persistLocalState(customProperties.getCacheKey(), currentState);
                } catch (Exception e) {
                    getLogger().info("Exception saving cache state processing flow file: " +
                            e.getMessage());
                }
                int available = 0;
                try {
                    available = customProperties.getContent().available();
                } catch (IOException e) {
                    // in case of the last part, the stream is already closed
                }
                getLogger().info("Success uploading part flowfile={} part={} available={} " +
                        "etag={} uploadId={}", customProperties.getFileName(), part, available,
                        uploadPartResult.getEtag(), currentState.getUploadId());
                if (available == 0) {
                    break;
                }
            } catch (Exception e) {
                getLogger().info("Failure uploading part flowfile={} part={} bucket={} key={} " +
                        "reason={}", customProperties.getFileName(), part, customProperties.getBucket(), customProperties.getObjectKey(), e.getMessage());
                throw (e);
            }
        }
    }

    // Set the Acl, with finer grained permission Settings taking precedence
    private void setAcl(CustomProperties customProperties, PutObjectBasicRequest request, ProcessContext context) {
        // Predefined permission policies have a low priority and may be overwritten
        final AccessControlList cannedAcl = createCannedACL(context, customProperties.getFlowFile());
        if (cannedAcl != null) {
            request.setAcl(cannedAcl);
        }
        final AccessControlList acl = createACL(context, customProperties.getFlowFile());
        if (acl != null) {
            request.setAcl(acl);
        }
    }

    private Map<String, Object> buildUserMetaData(final ProcessContext context, final CustomProperties customProperties) {
        final Map<String, Object> userMetadata = new HashMap<>();
        for (final Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet()) {
            if (entry.getKey().isDynamic()) {
                final String value = context.getProperty(
                        entry.getKey()).evaluateAttributeExpressions(customProperties.getFlowFile()).getValue();
                userMetadata.put(entry.getKey().getName(), value);
            }
        }
        return userMetadata;
    }

    private Map<String, String> getResultAttributes(PutResult putResult, ObjectMetadata objectMetadata) {
        Map<String, String> attributes = new HashMap<>();
        if (putResult.getVersionId() != null) {
            attributes.put(OBS_VERSION, putResult.getVersionId());
        }
        if (putResult.getETag() != null) {
            attributes.put(OBS_E_TAG, putResult.getETag());
        }
        attributes.put(OBS_STORAGE_CLASS, (putResult.getStorageClass() == null ? StorageClassEnum.STANDARD : putResult.getStorageClass()).getCode());
        if (objectMetadata.getAllMetadata().size() > 0) {
            StringBuilder userMetaBuilder = new StringBuilder();
            for (String userKey : objectMetadata.getAllMetadata().keySet()) {
                userMetaBuilder.append(userKey).append("=").append(objectMetadata.getAllMetadata().get(userKey));
            }
            attributes.put(OBS_USER_META, userMetaBuilder.toString());
        }
        attributes.put(OBS_API_METHOD_ATTR_KEY, putResult.getPutType());
        attributes.put(OBJECT_URL, putResult.getObjectUrl());
        return attributes;
    }

    private MultipartState getState(ObsClient obsClient, CustomProperties customProperties) throws IOException {
        try {
            MultipartState currentState = getLocalStateIfInOBS(obsClient, customProperties.getBucket(), customProperties.getCacheKey());

            if (currentState != null) {
                if (currentState.getPartETags().size() > 0) {
                    final PartEtag lastETag = currentState.getPartETags().get(
                            currentState.getPartETags().size() - 1);
                    getLogger().info("Resuming upload for flowfile='{}' bucket='{}' key='{}' " +
                                    "uploadID='{}' filePosition='{}' partSize='{}' storageClass='{}' " +
                                    "contentLength='{}' partsLoaded={} lastPart={}/{}",
                            customProperties.getFileName(), customProperties.getBucket(), customProperties.getObjectKey(), currentState.getUploadId(),
                            currentState.getFilePosition(), currentState.getPartSize(),
                            currentState.getStorageClass(),
                            currentState.getContentLength(),
                            currentState.getPartETags().size(),
                            Integer.toString(lastETag.getPartNumber()),
                            lastETag.getEtag());
                } else {
                    getLogger().info("Resuming upload for flowfile='{}' bucket='{}' key='{}' " +
                                    "uploadID='{}' filePosition='{}' partSize='{}' storageClass='{}' " +
                                    "contentLength='{}' no partsLoaded",
                            customProperties.getFileName(), customProperties.getBucket(), customProperties.getObjectKey(), currentState.getUploadId(),
                            currentState.getFilePosition(), currentState.getPartSize(),
                            currentState.getStorageClass().toString(),
                            currentState.getContentLength());
                }
            } else {
                currentState = new MultipartState();
                currentState.setPartSize(customProperties.getMultipartPartSize());
                currentState.setStorageClass(customProperties.getStorageClass());
                currentState.setContentLength(customProperties.getFlowFile().getSize());
                // Save the current upload status
                persistLocalState(customProperties.getCacheKey(), currentState);
                getLogger().info("Starting new upload for flowfile='{}' bucket='{}' key='{}'",
                        customProperties.getFileName(), customProperties.getBucket(), customProperties.getObjectKey());
            }
            return currentState;
        } catch (IOException e) {
            getLogger().error("IOException initiating cache state while processing flow files: " +
                    e.getMessage());
            throw (e);
        }

    }

    private final Lock bucketLock = new ReentrantLock();
    private final AtomicLong lastAgeOff = new AtomicLong(0L);
    private final DateFormat logFormat = new SimpleDateFormat();

    protected void ageOffUploads(final ProcessContext context, final ObsClient obsClient, final long now, String bucket) {
        MultipartUploadListing oldUploads = getAgeOffListAndAgeOffLocalState(context, obsClient, now, bucket);
        for (MultipartUpload upload : oldUploads.getMultipartTaskList()) {
            abortMultipartUpload(obsClient, oldUploads.getBucketName(), upload);
        }
    }

    public MultipartUploadListing getAgeOffListAndAgeOffLocalState(final ProcessContext context, final ObsClient obsClient, final long now, String bucket) {
        final long ageOffInterval = context.getProperty(MULTIPART_AGEOFF_INTERVAL).asTimePeriod(TimeUnit.MILLISECONDS);
        final Long maxAge = context.getProperty(MULTIPART_MAX_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
        final long ageCutOff = now - maxAge;

        final List<MultipartUpload> ageOffList = new ArrayList<>();
        // Uncompleted multisegment upload tasks are cleared every 60 minutes (by default)
        if ((lastAgeOff.get() < now - ageOffInterval) && bucketLock.tryLock()) {
            try {
                ListMultipartUploadsRequest listRequest = new ListMultipartUploadsRequest(bucket);
                MultipartUploadListing listing = obsClient.listMultipartUploads(listRequest);
                // All the multi-segment upload tasks under the bucket are traversed, and those that exceed the maximum threshold are recorded
                for (MultipartUpload upload : listing.getMultipartTaskList()) {
                    long firstUploadTime = upload.getInitiatedDate().getTime();
                    if (firstUploadTime < ageCutOff) {
                        ageOffList.add(upload);
                    }
                }
                // ageOff any local state
                ageOffLocalState(ageCutOff);
                // Update timestamp
                lastAgeOff.set(System.currentTimeMillis());
            } catch (Exception e) {
                if (e instanceof ObsException
                        && ((ObsException) e).getResponseCode() == 403
                        && ((ObsException) e).getErrorCode().equals("AccessDenied")) {
                    getLogger().warn("AccessDenied checking OBS Multipart Upload list for {}: {} " +
                                    "** The configured user does not have the OBS:ListBucketMultipartUploads permission " +
                                    "for this bucket, OBS ageoff cannot occur without this permission.  Next ageoff check " +
                                    "time is being advanced by interval to prevent checking on every upload **",
                            bucket, e.getMessage());
                    lastAgeOff.set(System.currentTimeMillis());
                } else {
                    getLogger().error("Error checking OBS Multipart Upload list for {}: {}",
                            bucket, e.getMessage());
                }
            } finally {
                bucketLock.unlock();
            }
        }
        return new MultipartUploadListing.Builder().bucketName(bucket).multipartTaskList(ageOffList).builder();
    }

    public void abortMultipartUpload(final ObsClient obsClient, final String bucket, final MultipartUpload upload) {
        final String uploadKey = upload.getObjectKey();
        final String uploadId = upload.getUploadId();
        final AbortMultipartUploadRequest abortRequest = new AbortMultipartUploadRequest(
                bucket, uploadKey, uploadId);
        // No call to an encryption service is necessary for an AbortMultipartUploadRequest.
        try {
            obsClient.abortMultipartUpload(abortRequest);
            getLogger().info("Aborting out of date multipart upload, bucket {} key {} ID {}, initiated {}",
                    bucket, uploadKey, uploadId, logFormat.format(upload.getInitiatedDate()));
        } catch (Exception ace) {
            getLogger().info("Error trying to abort multipart upload from bucket {} with key {} and ID {}: {}",
                    bucket, uploadKey, uploadId, ace.getMessage());
        }
    }

    public static class MultipartState implements Serializable {

        private static final long serialVersionUID = 9006072180563519740L;
        private String _uploadId;
        private Long _filePosition;
        private final List<PartEtag> _partETags;
        private Long _partSize;
        private StorageClassEnum _storageClassEnum;
        private Long _contentLength;
        private Long _timestamp;

        public MultipartState() {
            _uploadId = "";
            _filePosition = 0L;
            _partETags = new ArrayList<>();
            _partSize = 0L;
            _storageClassEnum = StorageClassEnum.STANDARD;
            _contentLength = 0L;
            _timestamp = System.currentTimeMillis();
        }

        public static MultipartState newMultipartState(String json) throws JsonProcessingException {
            JsonMapper jsonMapper = new JsonMapper();
            return jsonMapper.readValue(json, MultipartState.class);
        }

        public String getUploadId() {
            return _uploadId;
        }

        public void setUploadId(String id) {
            _uploadId = id;
        }

        public Long getFilePosition() {
            return _filePosition;
        }

        public void setFilePosition(Long pos) {
            _filePosition = pos;
        }

        public List<PartEtag> getPartETags() {
            return _partETags;
        }

        public void addPartETag(PartEtag tag) {
            _partETags.add(tag);
        }

        public Long getPartSize() {
            return _partSize;
        }

        public void setPartSize(Long size) {
            _partSize = size;
        }

        public StorageClassEnum getStorageClass() {
            return _storageClassEnum;
        }

        public void setStorageClass(StorageClassEnum storageClassEnum) {
            _storageClassEnum = storageClassEnum;
        }

        public Long getContentLength() {
            return _contentLength;
        }

        public void setContentLength(Long length) {
            _contentLength = length;
        }

        public Long getTimestamp() {
            return _timestamp;
        }

        public void setTimestamp(Long timestamp) {
            _timestamp = timestamp;
        }
    }

    public static class PutResult {
        private String bucketName;

        private String objectKey;

        private String eTag;

        private String versionId;

        private StorageClassEnum storageClass;

        private String objectUrl;

        private String location;

        private String encodingType;

        private InputStream callbackResponseBody;

        public String getETag() {
            return eTag;
        }

        public void setETag(String eTag) {
            this.eTag = eTag;
        }

        public String getPutType() {
            return putType;
        }

        public void setPutType(String putType) {
            this.putType = putType;
        }

        private String putType;

        public String getBucketName() {
            return bucketName;
        }

        public void setBucketName(String bucketName) {
            this.bucketName = bucketName;
        }

        public String getObjectKey() {
            return objectKey;
        }

        public void setObjectKey(String objectKey) {
            this.objectKey = objectKey;
        }

        public String getVersionId() {
            return versionId;
        }

        public void setVersionId(String versionId) {
            this.versionId = versionId;
        }

        public StorageClassEnum getStorageClass() {
            return storageClass;
        }

        public void setStorageClass(StorageClassEnum storageClass) {
            this.storageClass = storageClass;
        }

        public String getObjectUrl() {
            return objectUrl;
        }

        public void setObjectUrl(String objectUrl) {
            this.objectUrl = objectUrl;
        }

        public String getLocation() {
            return location;
        }

        public void setLocation(String location) {
            this.location = location;
        }

        public String getEncodingType() {
            return encodingType;
        }

        public void setEncodingType(String encodingType) {
            this.encodingType = encodingType;
        }

        public InputStream getCallbackResponseBody() {
            return callbackResponseBody;
        }

        public void setCallbackResponseBody(InputStream callbackResponseBody) {
            this.callbackResponseBody = callbackResponseBody;
        }
    }

    /**
     * 输入属性
     */
    public static class CustomProperties {
        private String bucket;
        private String objectKey;
        private String cacheKey;
        private String fileName;
        private Long multipartThreshold;
        private Long multipartPartSize;
        private String contentType;
        private String cacheControl;
        private ObsServiceEncryptionService encryptionService;
        private FlowFile flowFile;
        private InputStream content;
        private StorageClassEnum storageClass;

        public StorageClassEnum getStorageClass() {
            return storageClass;
        }

        public void setStorageClass(StorageClassEnum storageClass) {
            this.storageClass = storageClass;
        }

        public FlowFile getFlowFile() {
            return flowFile;
        }

        public void setFlowFile(FlowFile flowFile) {
            this.flowFile = flowFile;
        }

        public InputStream getContent() {
            return content;
        }

        public void setContent(InputStream content) {
            this.content = content;
        }

        public String getCacheKey() {
            return cacheKey;
        }

        public void setCacheKey(String cacheKey) {
            this.cacheKey = cacheKey;
        }

        public String getContentType() {
            return contentType;
        }

        public void setContentType(String contentType) {
            this.contentType = contentType;
        }

        public String getCacheControl() {
            return cacheControl;
        }

        public void setCacheControl(String cacheControl) {
            this.cacheControl = cacheControl;
        }

        public ObsServiceEncryptionService getEncryptionService() {
            return encryptionService;
        }

        public void setEncryptionService(ObsServiceEncryptionService encryptionService) {
            this.encryptionService = encryptionService;
        }

        public String getContentDisposition() {
            return contentDisposition;
        }

        public void setContentDisposition(String contentDisposition) {
            this.contentDisposition = contentDisposition;
        }

        private String contentDisposition;

        public String getBucket() {
            return bucket;
        }

        public void setBucket(String bucket) {
            this.bucket = bucket;
        }

        public String getObjectKey() {
            return objectKey;
        }

        public void setObjectKey(String objectKey) {
            this.objectKey = objectKey;
        }

        public String getFileName() {
            return fileName;
        }

        public void setFileName(String fileName) {
            this.fileName = fileName;
        }

        public Long getMultipartThreshold() {
            return multipartThreshold;
        }

        public void setMultipartThreshold(Long multipartThreshold) {
            this.multipartThreshold = multipartThreshold;
        }

        public Long getMultipartPartSize() {
            return multipartPartSize;
        }

        public void setMultipartPartSize(Long multipartPartSize) {
            this.multipartPartSize = multipartPartSize;
        }
    }
}
